package com.bigdata

import java.util.Properties

import com.bigdata.bean.{OrderEvent, OrderResult}
import com.bigdata.common.constants.Constants
import com.bigdata.common.utils.KafkaCommonUtils
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object OrderTimeoutDetect {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val source: DataStream[String] = KafkaCommonUtils.createKafkaSink(env, Constants.Topic_Order)

    val orderEventStream: DataStream[OrderEvent] = source.map{
      log =>
        val items: Array[String] = log.split(",")
        OrderEvent(items(0).toLong, items(1), items(2).toLong)
    }.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.milliseconds(1)) {
      override def extractTimestamp(element: OrderEvent): Long = element.eventTime
    })

//    val orderEventStream = env.fromCollection(List(
//      OrderEvent(1, "create", 1558430842000L),
//      OrderEvent(2, "create", 1558430843000L),
//      OrderEvent(2, "pay", 1558430844000L)
//    )).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.milliseconds(1)) {
//      override def extractTimestamp(element: OrderEvent): Long = element.eventTime
//    })

    // 定义一个带匹配时间窗口的模式
    val orderPayPattern: Pattern[OrderEvent, OrderEvent] = Pattern.begin[OrderEvent]("begin")
      .where(_.eventType == "create")
      .next("next")
      .where(_.eventType == "pay")
      .within(Time.minutes(5))

    // 定义一个输出标签
    val outputTag: OutputTag[OrderResult] = OutputTag[OrderResult]("orderTimeout")

    // 订单事件流根据 orderId 分流，然后在每一条流中匹配出定义好的模式
    val patternStream: PatternStream[OrderEvent] = CEP.pattern(orderEventStream.keyBy(_.orderId), orderPayPattern)

    import scala.collection.Map
    val complexResult: DataStream[OrderResult] = patternStream.select(outputTag)(
      // 对于已超时的部分模式匹配的事件序列，会调用这个函数
      (pattern: Map[String, Iterable[OrderEvent]], timeStamp: Long) => {
      val createOrder: Option[Iterable[OrderEvent]] = pattern.get("begin")
      OrderResult(createOrder.get.iterator.next().orderId, "timeout")
    })(
      // 检测到定义好的模式序列时，就会调用这个函数
      (pattern: Map[String, Iterable[OrderEvent]]) => {
      val payOrder: Option[Iterable[OrderEvent]] = pattern.get("next")
      OrderResult(payOrder.get.iterator.next().orderId, "success")
    })

    complexResult.print("complexResult").setParallelism(1)

    // 拿到同一输出标签中的 timeout 匹配结果（流）
    val timeoutResult: DataStream[OrderResult] = complexResult.getSideOutput(outputTag)
    timeoutResult.print("timeoutResult").setParallelism(1)

    env.execute("Order Timeout Detect Job")

  }
}
